package com.star.spring.reactor.flux;


import reactor.core.publisher.Flux;

/**
 * @description:
 * @author: star
 * @create: 2025-03-02 22:50
 **/

public class FluxGenerateDemo {

    public static void main(String[] args) throws InterruptedException {

//        1、just
//        Flux.just("a", "b", "c")
//                .subscribe(System.out::println);


//        2、interval
//        Flux.interval(Duration.ZERO, Duration.ofSeconds(1))
//                .subscribe(integer -> System.out.println("Received: " + integer));
//        Thread.sleep(20000);

//        3、range
//        Flux.range(1, 10)
//                .delayElements(Duration.ofSeconds(1))
//                .doOnNext(item -> System.out.println("next: " + item))
//
//                .map(item -> {
//                    if (item == 3) {
////                        throw new RuntimeException("error");
//                        // cancel流
//
//                    }
//                    return item;
//                })
//                .doOnComplete(() -> System.out.println("doOnComplete"))
//                .doOnError((e) -> System.out.println("doOnError"))
//                .subscribe();
//        Thread.sleep(10000);

//        4、generate
        Flux.generate(() -> 0, (state, sink) -> {

                    sink.next("3 * " + state + " = " + 3 * state);
                    if (state == 10) {
                        sink.complete();
                    }
                    if (state == 7) {
                        sink.error(new RuntimeException("error"));
                    }
                    return state + 1;
                })
                .doOnError(e -> System.out.println("onerror error"))
                .onErrorResume(e -> Flux.just("onerro resume：error"))
                .log()
                .subscribe((item) -> System.out.println("next: " + item)
                        , (e) -> System.out.println("subscripe error: " + e));



    }
}
